package com.atguigu.edu.realtime.dim.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.bean.TableProcessDim;
import com.atguigu.edu.realtime.common.util.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.*;

/**
 * 对关联后的数据进行处理函数类
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>> {
    //普通变量和算子状态变量都是为了存储配置信息
    //普通变量存放在内存中，状态可以保存在检查点中

    private MapStateDescriptor<String, TableProcessDim> mapStateDescriptor;

    public TableProcessFunction (MapStateDescriptor<String, TableProcessDim> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    //用来存放预加载的配置信息的普通变量
    private Map<String,TableProcessDim> configMap=new HashMap<>();
    //将配置表中的配置信息预加载到程序中
    @Override
    public void open(Configuration parameters) throws Exception {
        //从Mysql中读取配置信息


        Connection mysqlConnection = JdbcUtil.getMysqlConnection();
        String sql="select * from edu_realtime_config.dim_table_process";
        List<TableProcessDim> tableProcessDimList = JdbcUtil.queryList(mysqlConnection, sql,TableProcessDim.class,  true);
        for (TableProcessDim tableProcessDim : tableProcessDimList) {
            configMap.put(tableProcessDim.getSourceTable(), tableProcessDim);
        }
        JdbcUtil.closeConnection(mysqlConnection);

    }

    @Override //处理主流数据 根据表名到广播状态中获取对应的配置信息，如果能够获取到，说明处理的是维度数据，传递到下游
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.ReadOnlyContext readOnlyContext, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {

        //获取操作的业务数据库表的表名
        String table =jsonObject.getString("table");
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcessDim> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
        //根据表名到广播状态中获取对应的配置信息
        TableProcessDim tableProcessDim = null;
        //用短路的方法先用状态赋值！！！！
        if ((tableProcessDim = broadcastState.get(table))!=null || (tableProcessDim=configMap.get(table))!=null){
            //如果从广播状态中获取到了对应的配置，说明是维度数据，向下游传递
            JSONObject dataJsonObj  = jsonObject.getJSONObject("data");
            //再向下游传递数据前，过滤掉data中不需要传递的属性
            deletNotNeedColumns(dataJsonObj,tableProcessDim.getSinkColumns());
            //在向下游传递数据之前，补充操作类型
            String type = jsonObject.getString("type");
            dataJsonObj.put("type",type);
            collector.collect(Tuple2.of(dataJsonObj,tableProcessDim));
        }
    }



    @Override //处理广播流数据 将读取到的配置信息放到广播状态中 K:表名 V:实体类对象
    public void processBroadcastElement(TableProcessDim tableProcessDim, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.Context context, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
        //获取对配置表操作的类型
        String op=tableProcessDim.getOp();
        //获取广播状态
        BroadcastState<String, TableProcessDim> broadcastState = context.getBroadcastState(mapStateDescriptor);
        //获取业务数据维度信息表表名
        String sourceTable = tableProcessDim.getSourceTable();
        if("d".equals(op)){
            //从配置表中删除了一条配置信息 从广播状态中奖对应的配置删除掉
            broadcastState.remove(sourceTable);
            configMap.remove(sourceTable);
        }else{
            //不论是读取，添加还是更新，都要把配置表的最新配置放入广播状态中
            broadcastState.put(sourceTable,tableProcessDim);
            configMap.put(sourceTable,tableProcessDim);//这一步不做也行
        }
    }

    //从代码层面上来说，广播流数据和主流数据那个先来，哪个先处理，
    // 但是从业务层面来说，主流数据要从广播流状态中读取配置信息，如果读取不到，数据就会丢失（不往下传递）
    // 所以需要广播流先执行

    /**
     * 过滤掉主流数据中不需要向下传递的属性
     * @param dataJsonObj 主流中的data数据
     * @param sinkColumns 辅助信息的列数据
     */
    private static void deletNotNeedColumns(JSONObject dataJsonObj, String sinkColumns) {
        String[] columnArr  = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
/*        Iterator<Map.Entry<String, Object>> iterator = entrySet.iterator();
        for (;iterator.hasNext();) {
            Map.Entry<String, Object> entry=iterator.next();
            String key = entry.getKey();
            if(!columnList.contains(key)){
                iterator.remove();
            }
        }*/
        entrySet.removeIf(entry->!columnList.contains(entry.getKey()));


    }
}

